package com.showdor.springboot.kafka;

import com.showdor.springboot.ConsumerRecordDAO;
import com.showdor.springboot.SimpleRecordDAOProxy;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * @author showdor
 * @email chuanqi@outlook.com
 * @date 2023/7/5
 */
//@Component
public class BatchConsumerHandler {
    @Resource
    ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory2;
    @Value("${kafka.topic.name}")
    private String topicName;

    public void startListening(String topic) {
        ConsumerRecordDAO dao = new SimpleRecordDAOProxy();
        ConcurrentMessageListenerContainer<String, String> container = kafkaListenerContainerFactory2.createContainer(topic);
        container.setupMessageListener((BatchAcknowledgingMessageListener<String, String>) (records, ack) -> {
            dao.create(dao, records, ack);
        });
        //container.setConcurrency(5); // 设置并发性
        container.start();
    }

    @PostConstruct
    public void listener() {
        startListening(topicName);
    }

}
